package com.apache.log.es.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.apache.database.model.Page;
import com.apache.log.es.ESBaseDao;
import com.apache.log.es.config.EsClientFactory;
import com.apache.log.es.config.EsMap;
import com.apache.tools.StrUtil;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestResult;
import io.searchbox.core.*;
import io.searchbox.core.search.aggregation.DateHistogramAggregation;
import io.searchbox.core.search.aggregation.MaxAggregation;
import io.searchbox.core.search.aggregation.MetricAggregation;
import io.searchbox.core.search.aggregation.TermsAggregation;
import io.searchbox.indices.CreateIndex;
import io.searchbox.indices.IndicesExists;
import io.searchbox.indices.aliases.AddAliasMapping;
import io.searchbox.indices.aliases.ModifyAliases;
import org.elasticsearch.search.sort.SortBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;

/**
 * @description: ES统一操作DAO接口实现类
 * @author: Hou Dayu
 * @date: Created in 2020/11/17
 */
public class DefEsDaoImpl implements ESBaseDao {

    private Logger logger = LoggerFactory.getLogger("es logs");
    private JestClient jestClient;

    public DefEsDaoImpl(String jestHost){
        jestClient = EsClientFactory.buildEsClient(jestHost,0,0,0);
    }

    public DefEsDaoImpl(String jestHost,int maxTotal){
        jestClient = EsClientFactory.buildEsClient(jestHost,maxTotal,0,0);
    }

    public DefEsDaoImpl(String jestHost,int readTimeOut,int connTimeout){
        jestClient = EsClientFactory.buildEsClient(jestHost,0,readTimeOut,connTimeout);
    }
    /**
     * 构造函数
     * @param jestHost ES服务连接地址
     * @param readTimeOut 读取超时时间
     * @param connTimeout 连接超时时间
     * @param maxTotal 最大连接数
     */
    public DefEsDaoImpl(String jestHost,int readTimeOut,int connTimeout,int maxTotal){
        jestClient = EsClientFactory.buildEsClient(jestHost,maxTotal,readTimeOut,connTimeout);
    }

    @Override
    public JestClient getJestClient() {
        return jestClient;
    }

    @Override
    public boolean createIndex(String indexName, String type, Map<String, String> settings, String mapping) {
        try {
            if(!indexExists(indexName)) {
                String numReplicas = "0";//副本数
                String setting = null;
                if(null != settings && !settings.isEmpty()){
                    numReplicas = settings.get("numberOfReplicas");
                    String numbershards = StrUtil.doNull(settings.get("numbershards"),"5");//分片数
                    setting = "{\"index.number_of_shards\" : "+numbershards+",\"number_of_replicas\": "+numReplicas+",\"index.refresh_interval\": \"30s\"}";
                } else {
                    setting = "{\"index.number_of_shards\" : 5,\"number_of_replicas\" : "+numReplicas + ",\"index.refresh_interval\": \"30s\"}";
                }
                logger.info("创建索引{},setting={},mappings={}",indexName,setting,mapping.toString());
                JestResult result = jestClient.execute(new CreateIndex.Builder(indexName.toLowerCase()).settings(setting).mappings(mapping.toString()).build());
                if (result == null || !result.isSucceeded()) {
                    logger.error("创建索引失败：{}",result);
                    return false;
                }
                if(null != settings && settings.containsKey("aliases")) {//别名
                    AddAliasMapping build = new AddAliasMapping.Builder(indexName, settings.get("aliases")).build();
                    JestResult jestResult = jestClient.execute(new ModifyAliases.Builder(build).build());
                    logger.info("创建索引结果结果：{}",jestResult);
                }
                //创建索引
                logger.info("创建索引[{}]结果为{}",indexName,result.getJsonString());
                return result.isSucceeded();
            }
            return true;
        } catch(Exception e){
            logger.error("创建索引失败,indexName={},error={}",indexName,e.getMessage());
            return false;
        }
    }

    @Override
    public boolean indexExists(String indexName) {
        try {
            JestResult result = jestClient.execute(new IndicesExists.Builder(indexName.toLowerCase()).build());
            if (result==null || !result.isSucceeded()) {
                return false;
            }
        } catch (Exception e) {
            return false;
        }
        return true;
    }

    @Override
    public boolean createIndexInfo(String indexName, String type, String docId, Map<String, Object> json) {
        try {
            Index.Builder indexBuilder=new Index.Builder(json);
            indexBuilder.index(indexName).type(type).id(String.valueOf(docId));
            Index index = indexBuilder.build();
            DocumentResult result = jestClient.execute(index);
            if (result==null || !result.isSucceeded()) {
                return false;
            }
            logger.info("create index info result {}",result.getJsonString());
        } catch (IOException e) {
            logger.info("添加索引文档失败,error={}",e.getMessage());
            return false;
        }
        return true;
    }

    @Override
    public boolean createIndexBatch(String indexName, String type, List<Map<String, Object>> jsons) {
        if(StrUtil.isEmpty(jsons)){
            return false;
        }
        int count = jsons.size();
        try {
            List<Index> monitorList = new ArrayList();
            for(int i=0; i<count; i++){
                Map<String,Object> info = jsons.get(i);
                String id = StrUtil.doNull(String.valueOf(info.get("rowkey")),"");
                monitorList.add(new Index.Builder(info).id(id).build());
            }
            Bulk bulk = new Bulk.Builder().defaultIndex(indexName).defaultType("doc")
                    .addAction(monitorList).build();
            jestClient.execute(bulk);
            monitorList = null;
        } catch (IOException e) {
            return false;
        }
        return true;
    }

    @Override
    public boolean updateIndexInfo(String indexName, String type, String docId, Map<String, Object> json) {
        try {
            Update index1 = new Update.Builder(json).index(indexName).id(docId).build();
            JestResult result = jestClient.execute(index1);
            if (result==null || !result.isSucceeded()) {
                return false;
            }
        } catch (Exception e) {
            return false;
        }
        return true;
    }

    @Override
    public boolean delIndexInfo(String indexName, String docId) {
        try {
            Delete delete = new Delete.Builder(docId).index(indexName).build();
            JestResult result = jestClient.execute(delete);
            if (result==null || !result.isSucceeded()) {
                return false;
            }
        } catch (Exception e) {
            return false;
        }
        return true;
    }

    @Override
    public Page pageList(String indexName, String indexType, int pageIndex, int pageSize, String query, SortBuilder sortBuilder) {
        pageIndex = (pageIndex == 0) ? 1 : pageIndex;
        pageSize = (pageSize == 0) ? 5 : pageSize;
        Page page = new Page();
        page.setPageNeeded(pageIndex);
        page.setPageSize(pageSize);
        if (StrUtil.isNull(query)) {
            logger.warn("缺失查询参数");
            return page;
        }
        int from = (pageIndex - 1) * pageSize;//设置起始页
        JSONObject json = new JSONObject();
        json.put("from", from);
        json.put("size", pageSize);
        json.put("query", JSON.parseObject(query));
        if (null != sortBuilder) {//排序设置
            json.put("sort", JSON.parseObject(sortBuilder.toString().replaceAll("\n","")));
        }
        SearchResult searchResult = this.execute(jestClient,"pageList", indexName,indexType,json.toString(),"");
        if (searchResult == null || !searchResult.isSucceeded()) {
            logger.warn("查询出错:{}",searchResult);
            return page;
        }
        Long count = searchResult.getTotal();
        if (count == 0) {
            return page;
        }
        List hits = searchResult.getHits(EsMap.class);
        int pageCount = (int) Math.ceil(count/pageSize);
        page.setPageCount(pageCount);
        page.setCount(count.intValue());
        List list = new ArrayList();
        for (int i = 0; i < hits.size(); i++) {
            SearchResult.Hit<EsMap, Void> hit = (SearchResult.Hit<EsMap, Void>) hits.get(i);
            list.add(hit.source);
        }
        page.setPageObjects(list);
        return page;
    }

    @Override
    public Map<String, Object> pageList(String indexName, String infoType, String queryStr) {
        Map<String,Object> data = new HashMap();
        data.put("total","0");
        List list = new ArrayList();
        SearchResult searchResult = this.execute(jestClient,"pageList", indexName,infoType,queryStr,"");
        if (searchResult == null || !searchResult.isSucceeded() || searchResult.getTotal()==0) {
            return data;
        }
        List hits = searchResult.getHits(EsMap.class);
        for (int i = 0; i < hits.size(); i++) {
            SearchResult.Hit<EsMap, Void> hit = (SearchResult.Hit<EsMap, Void>) hits.get(i);
            list.add(hit.source);
        }
        data.put("total",searchResult.getTotal());
        data.put("rows",list);
        return data;
    }

    @Override
    public Map selectById(String indexName, String id) {
        Get get = new Get.Builder(indexName, id).build();
        try {
            JestResult result = jestClient.execute(get);
            Map data = new HashMap();
            if(result.isSucceeded()) {
                data = result.getSourceAsObject(Map.class);
            }
            return data;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public Object searchByQuery(String indexName, String infoType, String query) {
        SearchResult searchResult = this.execute(jestClient,"searchByQuery", indexName,infoType,query,"");
        if (searchResult == null || !searchResult.isSucceeded()) {
            return null;
        }
        Long count = searchResult.getTotal();
        if (count == 0) {
            return null;
        }
        MetricAggregation ma = searchResult.getAggregations();
        JSONObject aggs = JSON.parseObject(query).getJSONObject("aggs");
        if(aggs == null || aggs.size()==0){
            return searchResult.getTotal();
        }
        Iterator<String> it = aggs.keySet().iterator();
        String key = null;
        while(it.hasNext()){
            key = it.next();
            break;
        }
        try {
            DateHistogramAggregation dd = ma.getDateHistogramAggregation(key);
            List<DateHistogramAggregation.DateHistogram> list = dd.getBuckets();
            List<Object> data = new ArrayList();
            for (DateHistogramAggregation.DateHistogram dh : list) {
                net.sf.json.JSONObject dataObj = net.sf.json.JSONObject.fromObject(dh);
                MaxAggregation max = dh.getMaxAggregation("1");
                if (!StrUtil.isEmpty(max)) {
                    dataObj.put("value", max.getMax());
                }
                TermsAggregation terms = dh.getTermsAggregation("3");
                if (null != terms) {
                    dataObj.put("value", terms.getBuckets());
                }
                data.add(dataObj);
            }
            return data;
        } catch (Exception e){
            TermsAggregation value = ma.getTermsAggregation(key);
            return value.getBuckets();
        }
    }

    private SearchResult execute(JestClient client, String methodName, String indexName, String indexType, String searchStr, String traceId){
        Long time = System.currentTimeMillis();
        SearchResult result = null;
        /**
         * _primary: 指查询只在主分片中查询
         * _primary_first: 指查询会先在主分片中查询，如果主分片找不到（挂了），就会在副本中查询。
         * _local: 指查询操作会优先在本地节点有的分片中查询，没有的话再在其它节点查询。
         */
        Search search = new Search.Builder(searchStr).addIndex(indexName)
                .setParameter("preference","_primary").build();
        try {
            result = client.execute(search);
        }catch(Exception e){
            logger.error(" traceid = {}; ES-search[{}] Exception={}",traceId,methodName, e.getMessage());
        } finally {
            logger.info(" traceid = {}; esCost = {}; appName = esReq; esRequestBody = {}",traceId,(System.currentTimeMillis() - time),searchStr);
        }
        return result;
    }

    private <T> List<T> execute(JestClient client, Class<T> sourceType, String methodName, String indexName, String indexType, JSONObject queryJson, int from, int pageSize){
        Long time = System.currentTimeMillis();
        JSONObject json = new JSONObject();
        json.put("from", from);
        json.put("size", pageSize);
        json.put("query", queryJson);
        Search search = new Search.Builder(json.toString()).addIndex(indexName).build();
        try {
            JestResult result = client.execute(search);
            if (result != null && result.isSucceeded()) {
                List list = result.getSourceAsObjectList(sourceType);
                return list;
            }
        } catch (IOException e) {
            logger.error("ES查询["+methodName+"]出错",e);
        } finally {
            logger.info(" traceid = {}; esCost = {}; appName = esReq; esRequestBody = {}",time+"",(System.currentTimeMillis() - time),json.toString());
        }
        return new ArrayList(1);
    }
}
